package com.scheduler.actor.server;

import com.alibaba.fastjson.JSON;
import com.scheduler.actor.ActorMessage;
import com.scheduler.actor.WorkerActorEventBus;
import com.scheduler.utils.trace.Slf4jMdcUtil;
import io.vertx.core.Vertx;
import io.vertx.core.VertxOptions;
import io.vertx.core.http.HttpServerOptions;
import io.vertx.core.http.HttpServerRequest;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.MDC;

import java.util.Objects;

@Slf4j
public class VertxServer {

    private WorkerActorEventBus workerActorEventBus;

    public VertxServer(WorkerActorEventBus workerActorEventBus) {
        this.workerActorEventBus = workerActorEventBus;
    }

    /**
     * 如果服务启动了，但无法访问，通常是异常了，通常的异常原因是Netty版本不对。
     * 无需引入vertx-web，只有简单两个路由，手动处理
     */
    public void startServer(int port) {
        log.info("初始化VertxServer开始，如果长时间未成功，检查一下Netty版本是否有冲突");
        VertxOptions vertxOptions = new VertxOptions();
        // 改一下线程池大小，不需要很大，因为任务都在线程池处理的
        vertxOptions.setEventLoopPoolSize(1);
        vertxOptions.setWorkerPoolSize(VertxOptions.DEFAULT_WORKER_POOL_SIZE);
        Vertx vertx = Vertx.vertx(vertxOptions);
        HttpServerOptions httpServerOptions = new HttpServerOptions();
        // 先用默认值，不做配置
        vertx.createHttpServer(httpServerOptions)
                .exceptionHandler(t -> log.error("异常，检查一下业务逻辑或Netty版本", t))
                .invalidRequestHandler(request -> log.error("不合法请求？这是什么东西，检查一下"))
                .requestHandler(this::doSomething)
                .listen(port, res -> {
                    if (res.succeeded()) {
                        log.info("Server is now listening!");
                    } else {
                        log.info("Failed to bind!");
                    }
                });
        log.info("初始化VertxServer完成，监听端口{}", port);

    }

    private void doSomething(HttpServerRequest request) {
        String sourceIp = request.getHeader("sourceIp");
        String traceId = request.getHeader("traceId");
        String uri = request.uri();
        request.bodyHandler(event -> {
            try {
                Slf4jMdcUtil.appendMDCWithTraceId(traceId);
                MDC.put("traceId", traceId);
                String result = dispatch(uri, sourceIp, event.toString());
                request.response()
                        .putHeader("content-type", "text/plain")
                        .end(result);
            } catch (Throwable t) {
                log.error("异常，检查一下业务逻辑2", t);
                request.response()
                        .putHeader("content-type", "text/plain")
                        .send("FAIL");
            } finally {
                Slf4jMdcUtil.clearMDCAndComplete();
            }
        });
    }

    private String dispatch(String uri, String sourceIp, String body) {
        if (!Objects.equals(workerActorEventBus.getRpcUrl(), uri)) {
            log.info("uri:{},不正确", uri);
            return "UNKNOWN";
        }
        if (body != null && !body.contains("PING")) {
            log.info("接收到任务分发：sourceIp={}, uri={}, body={}", sourceIp, uri, body);
        }
        ActorMessage actorMessage = JSON.parseObject(body, ActorMessage.class);
        return Objects.toString(workerActorEventBus.receiptTask(actorMessage), null);
    }
}
